package com.xbai.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * RDD CheckPoint
  * Spark 中对于数据的保存除了持久化操作之外，还提供了一种检查点的机制，
  * 检查点（本质是通过将 RDD 写入 Disk 做检查点）是为了通过 lineage 做容错的辅助，lineage 过长会造成容错成本过高，
  * 这样就不如在中间阶段做检查点容错，如果之后有节点出现问题而丢失分区，从做检查点的RDD开始重做 Lineage，就会减少开销。
  * 检查点通过将数据写入到 HDFS 文件系统实现了 RDD 的检查点功能。
  *
  * 为当前 RDD 设置检查点。该函数将会创建一个二进制的文件，并存储到 checkpoint 目录中，该目录是用 SparkContext.setCheckpointDir() 设置的。
  * 在 checkpoint 的过程中，该 RDD 的所有依赖于父 RDD 中的信息将全部被移除。
  * 对 RDD 进行 checkpoint 操作并不会马上被执行，必须执行 Action操作才能触发。
  *
  * @author xbai
  * @Date 2021/1/2
  */
object Spark_CheckPoint {

  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("CheckPoint")
    val sc = new SparkContext(conf)

    sc.setCheckpointDir("cp")
    val rdd: RDD[String] = sc.makeRDD(Array("xbai"))
    val ch: RDD[String] = rdd.map(_.toString + System.currentTimeMillis())
    ch.checkpoint()
    ch.foreach(println)
    ch.foreach(println)
    ch.foreach(println)
    ch.foreach(println)
  }
}
